package com.song.elasticsearch.config.kafka;

import lombok.Data;
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.config.SaslConfigs;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.Properties;

@Data
@Configuration
@ConditionalOnBean(KafkaSecurity.class)
@ConfigurationProperties(prefix = "kafka.producer")
public class KafkaProducerConfig {
    private String bootstrapServers;
    private String keySerializer;
    private String valueSerializer;
    private String securityProtocol;
    private String saslKerberosServiceName;
    private String kerberosDomainName;

    private Boolean producerSign = false;

    @Autowired
    KafkaSecurity kafkaSecurity;

    @Bean("kafkaProducer")
    public KafkaProducer<String, String> kafkaProducer(){
        if (!producerSign){
            kafkaSecurity.securityValidate();
            producerSign = true;
        }
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(producerProperties());
        return kafkaProducer;
    }

    public Properties producerProperties(){
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, keySerializer);
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, valueSerializer);
        properties.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, securityProtocol);
        // 服务名
        properties.put(SaslConfigs.SASL_KERBEROS_SERVICE_NAME, saslKerberosServiceName);
        // 域名
        properties.put(SaslConfigs.KERBEROS_DOMAIN_NAME, kerberosDomainName);
        return properties;
    }
}
